package com.yl.websocket;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.yl.domain.ConferenceApplyDO;
import com.yl.mapper.ConferenceApplyMapper;
import com.yl.service.ConferenceCourseService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

/**
 * websocket 消息处理
 *
 * @author ruoyi
 */
@Component
@ServerEndpoint("/websocket/message/{conferenceId}/{userId}")
public class WebSocketServer {


    private static ApplicationContext applicationContext;

    //解决无法注入 mapper 问题
    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketServer.applicationContext = applicationContext;
    }

    /**
     * WebSocketServer 日志控制器
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);


    /**
     * 用户集 key 是会议号   session 是用户
     */
    private static Map<Session, Long> confernecenMap = new ConcurrentHashMap<Session, Long>();


    /**
     * 默认最多允许同时在线人数100
     */
    public static int socketMaxOnlineCount = 10000;

    private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") Long userId, @PathParam("conferenceId") Long conferenceId) throws Exception {
        boolean semaphoreFlag = false;
        // 尝试获取信号量
        semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
        if (!semaphoreFlag) {
            // 未获取到信号量
            LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
            WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
            session.close();
        } else {
            // 添加用户
            confernecenMap.put(session, conferenceId);
            WebSocketUsers.put(userId, conferenceId, session);
            LOGGER.info("\n 建立连接 - {},key - {},频道号 - {},当前人数 - {}", session, userId, conferenceId, WebSocketUsers.getUsers(conferenceId).size());
            // WebSocketUsers.sendMessageToUserByText(session, "连接成功");
        }
    }


    /**
     * 连接关闭时处理
     */
    @OnClose
    public void onClose(Session session) {
        LOGGER.info("\n 关闭连接 - {}", session);
        Long aLong = confernecenMap.get(session);
        // 移除用户
        WebSocketUsers.remove(session, aLong);
        // 获取到信号量则需释放
        SemaphoreUtils.release(socketSemaphore);
    }

    /**
     * 抛出异常时处理
     */
    @OnError
    public void onError(Session session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            // 关闭连接
            session.close();
        }
        String sessionId = session.getId();
        LOGGER.info("\n 连接异常 - {}", sessionId);
        LOGGER.info("\n 异常信息 - {}", exception);
        Long aLong = confernecenMap.get(session);
        // 移出用户
        WebSocketUsers.remove(session, aLong);
        // 获取到信号量则需释放
        SemaphoreUtils.release(socketSemaphore);
    }

    /**
     * 服务器接收到客户端消息时调用的方法
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // String msg = message.replace("你", "我").replace("吗", "");
        LOGGER.info("onMessage 接收到的消息为 - {}", message);
        if (StringUtils.isNotBlank(message)) {
            if (message.equals("ping")) {
                // 心跳
                System.out.println("心跳回复-------");
                WebSocketUsers.sendMessageToUserByText(session, message);
                return;
            }
            Map parseObject = JSONObject.parseObject(message, Map.class);
            Long conferenceId = Long.valueOf(parseObject.get("conferenceId").toString());
            ConferenceApplyMapper conferenceApplyMapper = applicationContext.getBean(ConferenceApplyMapper.class);
            if (parseObject.containsKey("level")) {
                // 存上会议过程 用于前端调用
                ConferenceCourseService conferenceCourseService = applicationContext.getBean(ConferenceCourseService.class);
                conferenceCourseService.websocketSave(conferenceId, JSON.toJSONString(parseObject));
                if (parseObject.get("level").equals("0")) {
                    LOGGER.info("准备阶段========会议id - {}. param - {}", conferenceId, parseObject);
                    // 0是准备阶段 ，谁准备了往房间里面群发一条 已准备的列表，并且记录下来
                    Long keyBySession = WebSocketUsers.getKeyBySession(conferenceId, session);
                    conferenceApplyMapper.prepareByConferenceId(conferenceId, keyBySession);
                    List<ConferenceApplyDO> list = conferenceApplyMapper.selectList(Wrappers.<ConferenceApplyDO>lambdaQuery()
                            .eq(ConferenceApplyDO::getUnJoin, "1")
                            .eq(ConferenceApplyDO::getConferenceId, conferenceId));
                    if (list != null && list.size() != 0) {
                        parseObject.put("studentList", list.stream().collect(Collectors.toList()));
                        WebSocketUsers.sendAllToUsersByText(JSON.toJSONString(parseObject), conferenceId);
                        return;
                    }
                } else if (parseObject.get("level").equals("1")) {
                    LOGGER.info("第一阶段开始========会议id - {}. param - {}", conferenceId, parseObject);
                    List<ConferenceApplyDO> list = conferenceApplyMapper.selectList(Wrappers.<ConferenceApplyDO>lambdaQuery()
                            .eq(ConferenceApplyDO::getUnJoin, "1")
                            .eq(ConferenceApplyDO::getConferenceId, conferenceId));
                    if (list != null && list.size() != 0) {
                        parseObject.put("studentList", list.stream().collect(Collectors.toList()));
                        WebSocketUsers.sendAllToUsersByText(JSON.toJSONString(parseObject), conferenceId);
                        return;
                    }
                } else if (parseObject.get("level").equals("2")) {
                    LOGGER.info("第二阶段开始---------会议id - {}, param -{}",conferenceId,parseObject);
                    //  第二阶段开始 准备倒计时
                    if (parseObject.get("secondBegin").equals("1")) {
                        // 第二环节总时长
                        String stageNum = parseObject.get("secondStageNum").toString();
                        System.out.println("stageNum=" + stageNum);
                        Integer secondStageNum = Integer.valueOf(stageNum);
                        // todo 自定义10秒
                        daojishi(secondStageNum, conferenceId);
                        parseObject.put("secondSpeakTime", System.currentTimeMillis());
                        parseObject.put("secondBegin", "0");
                    }
                    WebSocketUsers.sendAllToUsersByText(JSON.toJSONString(parseObject), conferenceId);
                } else if (parseObject.get("level").equals("3")) {
                    LOGGER.info("第三阶段开始========会议id - {}. param - {}", conferenceId, parseObject);
                    // 第三阶段开始了
                    if(StringUtils.isNotBlank(parseObject.get("voteId").toString())){
                        LOGGER.info("第三阶段投票阶段开始---会议id - {}. param - {}",conferenceId,parseObject);
                        // 选举人+1
                        Long voteId = Long.valueOf(parseObject.get("voteId").toString());
                        // 发言人id
                        Long userId = Long.valueOf(parseObject.get("userId").toString());
                        // 投谁了 谁+1
                        conferenceApplyMapper.voteAdd(conferenceId, voteId);
                        // 标记投票人是否已投票
                        conferenceApplyMapper.updateVote(conferenceId, userId);
                        List<ConferenceApplyDO> list = conferenceApplyMapper.selectList(Wrappers.<ConferenceApplyDO>lambdaQuery()
                                .eq(ConferenceApplyDO::getUnJoin, "1")
                                .eq(ConferenceApplyDO::getConferenceId, conferenceId));
                        parseObject.put("studentList", list.stream().collect(Collectors.toList()));
                        List<Long> collect = list.stream().filter(f -> f.getUnVote().equals("1")).map(ConferenceApplyDO::getId).collect(Collectors.toList());
                        // 已投票人数据
                        parseObject.put("voteIdArr", collect == null ? new ArrayList<>() : collect);
                        parseObject.put("voteId","");
                        parseObject.put("userId","");
                    }
                    WebSocketUsers.sendAllToUsersByText(JSON.toJSONString(parseObject), conferenceId);
                } else if (parseObject.get("level").equals("4")) {
                    LOGGER.info("会议结束了========会议id - {}. 参数 - {}", conferenceId, parseObject);
                    // 会议结束了
                    WebSocketUsers.sendAllToUsersByText(JSON.toJSONString(parseObject), conferenceId);
                }
            } //level不为空
        }
    }

    /**
     * 第二阶段倒计时
     *
     * @author : y
     * @date :  2024/7/16 22:39
     **/
    public void daojishi(Integer miao, Long conferenceId) {
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        LOGGER.info("第二阶段倒计时开始了！当前时间 time - {}, 倒计时分钟 min - {},会议id - {}",System.currentTimeMillis(),miao,conferenceId);
        // 设置倒计时的时间，比如这里设置为10秒
        int countdownSeconds = (miao * 60);
        // 提交一个任务，延迟countdownSeconds秒后执行
        executor.schedule(() -> {
            // 这里可以添加倒计时结束后需要执行的代码
            ConferenceApplyMapper conferenceApplyMapper = applicationContext.getBean(ConferenceApplyMapper.class);
            List<ConferenceApplyDO> list = conferenceApplyMapper.selectList(Wrappers.<ConferenceApplyDO>lambdaQuery()
                    .eq(ConferenceApplyDO::getUnJoin, "1")
                    .eq(ConferenceApplyDO::getConferenceId, conferenceId));
            HashMap<String, Object> map = new HashMap<>(16);
            map.put("conferenceId", conferenceId.toString());
            map.put("repId", "");
            map.put("repNo", "");
            // //当前环节 0准备环节 1 轮流发言环节 2 抢答发言环节 3 总结发言环节
            map.put("level", "3");
            map.put("userId", "");
            map.put("voteId", "");
            map.put("secondBegin", "");
            // //第二环节总时长
            map.put("secondStageNum", "");
            // //第二环节开始时间戳
            map.put("secondSpeakTime", System.currentTimeMillis());
            // //第三环节投票倒计时时间戳
            map.put("thirdSpeakTime", System.currentTimeMillis());
            map.put("studentList", list);
            LOGGER.info("第二阶段倒计时结束 ----群发内容 [第三阶段开始了]===={}", map);
            ConferenceCourseService conferenceCourseService = applicationContext.getBean(ConferenceCourseService.class);
            conferenceCourseService.websocketSave(conferenceId, JSON.toJSONString(map));
            WebSocketUsers.sendAllToUsersByText(JSON.toJSONString(map), conferenceId);
        }, countdownSeconds, TimeUnit.SECONDS);
        // 关闭ScheduledExecutorService，防止资源泄露
        executor.shutdown();
    }
}
